package cn.hy.file.read;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.LineIterator;

import java.io.File;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;


/**
 * 大文件读写
 */
@Slf4j
public class BigFileHandler {

    private final static ExecutorService executorService = new ThreadPoolExecutor(
            10,
            20,
            30,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(600),
            new ThreadFactoryBuilder().setNameFormat("handle-task-%d").build(),
            new ThreadPoolExecutor.CallerRunsPolicy());


    /**
     * 切分大文件
     * split -l 100000 largeFile.txt -d -a 4 smallFile_
     * @param file file
     */
    public void handle(File file) {
        // 流式读取的方式，一行一行的读取数据。
        // 方案2
        // Files#lines底层源码，其实原理跟上面的 LineIterator类似，同样也是封装成迭代器模式。
//        Files.lines(Paths.get("文件路径"), Charset.defaultCharset()).forEach(line -> {
//            convertToDB(line);
//        });
        try (LineIterator iterator = IOUtils.lineIterator(Files.newInputStream(file.toPath()), "UTF-8")) {
            // 存储每个任务执行的行数
            List<String> lines = new ArrayList<>();
            // 存储异步任务
            List<ConvertTask> tasks = new ArrayList<>();
            while (iterator.hasNext()) {
                String line = iterator.nextLine();
                lines.add(line);
                // 设置每个线程执行的行数
                if (lines.size() == 1000) {
                    // 新建异步任务，注意这里需要创建一个 List
                    tasks.add(new ConvertTask(new ArrayList<>(lines)));
                    lines.clear();
                }
                if (tasks.size() == 10) {
                    asyncBatchExecuteTask(tasks);
                }
            }
            // 文件读取结束，但是可能还存在未被内容
            tasks.add(new ConvertTask(new ArrayList<>(lines)));
            // 最后再执行一次
            asyncBatchExecuteTask(tasks);
        } catch (Exception e) {
            log.error("handle error ", e);
        }
    }

    /**
     * CountDownLatch 批量执行
     */
    private static void asyncBatchExecuteTask(List<ConvertTask> tasks) {
        CountDownLatch countDownLatch = new CountDownLatch(tasks.size());
        for (ConvertTask task : tasks) {
            task.setCountDownLatch(countDownLatch);
            executorService.submit(task);
        }
        // 主线程等待异步线程 countDownLatch 执行结束
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
           Thread.currentThread().interrupt();
           log.info("interrupt",e);
        }
        // 清空，重新添加任务
        tasks.clear();
    }


    /**
     * 异步任务
     * 等数据导入完成之后，一定要调用 countDownLatch.countDown()
     * 不然，这个主线程将会被阻塞，
     */
    private static class ConvertTask implements Runnable {

        private CountDownLatch countDownLatch;

        private final List<String> lines;

        public ConvertTask(List<String> lines) {
            this.lines = lines;
        }

        public void setCountDownLatch(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                for (String line : lines) {
                    // convertToDB(line);
                }
            } finally {
                countDownLatch.countDown();
            }
        }
    }
}
